package com.publisherConfirm;

import com.constant.Constants;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

//发布确认
public class PublishConfirm {
    private static final Integer MESSAGE_COUNT = 10000;
        //1.建立连接,提取出来
    public static Connection createConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.POST);
        connectionFactory.setUsername(Constants.USER_NAME);
        connectionFactory.setPassword(Constants.PASSWORD);
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        return connectionFactory.newConnection();
    }
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        //1.单独确认
//        publishingMessageIndividually();
        //2.批量确认
        publishingMessageINBatches();
        //3.异步确认
        handlingPublisherConfirmAsynchronously();

    }
    //异步确认
    private static void handlingPublisherConfirmAsynchronously() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()){
            //2.开启信道
            Channel channel = connection.createChannel();
            //设置信道模式 为confirm模式
            channel.confirmSelect();
            //3.创建交换机,使用内置交换机
            //4.声明队列:
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3,true,false,false,null);
            //监听confirm
            //5.发送消息,并等待确认
            long start = System.currentTimeMillis();
            //集合中未确认的消息ID
            SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());

            channel.addConfirmListener(new ConfirmListener() {
                //确认应答:
                @Override   //当前发送的消息数    是否批量响应
                public void handleAck(long l, boolean b) throws IOException {
                    if(b){
                        //批量响应 ,
                        confirmSeqNo.headSet(l+1).clear();
                    }else{
                        //仅处理当前条消息
                        confirmSeqNo.remove(l);
                    }
                }
                //不确认应答:
                @Override
                public void handleNack(long l, boolean b) throws IOException {
                    if(b){
                        confirmSeqNo.headSet(l+1).clear();
                    }else{
                        confirmSeqNo.remove(l);
                    }
                }
            });
            //发送消息
            for(int i=0;i<MESSAGE_COUNT;i++){
                //构造消息,并发送
                String msg = "publish confirm..."+i;
                long seqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE3,null,msg.getBytes());
                confirmSeqNo.add(seqNo);
            }
            //当集合中的消息未处理完时,进行阻塞等待
            while(!confirmSeqNo.isEmpty()){
                Thread.sleep(10);
            }
            long end = System.currentTimeMillis();
            System.out.printf("异步确认策略,消息条数: %d ,耗时: %d ms \n",MESSAGE_COUNT,end-start);
        }
    }
    //批量确认
    private static void publishingMessageINBatches() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()){
            //2.开启信道
            Channel channel = connection.createChannel();
            //设置信道模式 为confirm模式
            channel.confirmSelect();
            //3.创建交换机,使用内置交换机
            //4.声明队列:
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2,true,false,false,null);
            //5.发送消息,并等待确认
            long start = System.currentTimeMillis();
            int batchSize = 100;//一次批量处理100条消息
            int outMessageCount = 0;//发出的消息条数
            for(int i=0;i<MESSAGE_COUNT;i++){
                //构造消息,并发送
                String msg = "publish confirm..."+i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE2,null,msg.getBytes());
                outMessageCount++;
                //当发出的消息条数够的时候,就进行确认一批
                if(outMessageCount==batchSize){
                    channel.waitForConfirmsOrDie(5000);
                    //再次将其设置为0,从新计数
                    outMessageCount = 0;
                }
            }
            if(outMessageCount>0){
                //若还有剩余的消息为响应,再次确认
                channel.waitForConfirmsOrDie(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("批量确认策略,消息条数: %d ,耗时: %d ms \n",MESSAGE_COUNT,end-start);
        }
    }

    //1.单独确认
    private static void publishingMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection = createConnection()){
            //2.开启信道
            Channel channel = connection.createChannel();
            //设置信道模式 为confirm模式
            channel.confirmSelect();
            //3.创建交换机,使用内置交换机
            //4.声明队列:
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1,true,false,false,null);
            //5.发送消息,并等待确认
            long start = System.currentTimeMillis();
            for(int i=0;i<MESSAGE_COUNT;i++){
                //构造消息,并发送
                String msg = "publish confirm..."+i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE1,null,msg.getBytes());
                //等待确认
                channel.waitForConfirmsOrDie(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("单独确认策略,消息条数: %d ,耗时: %d ms \n",MESSAGE_COUNT,end-start);
        }
    }
}
